#include <zmq.h>
#include <stdio.h>
#include <stdlib.h>
#include <glib.h>
#include <string.h>
#include "ipc.h"

#define IPC_PREFIX "ipc:///tmp/"
void *_ipc_context = NULL;
int exiting = 0;
GData *fmap;
GData *dmap;
GData *smap;
GThreadPool *sig_pool;
char *daemon_name;

G_LOCK_DEFINE(ipc_handler);

gpointer ipc_handler(gpointer data){
	zmq_msg_t request;
	zmq_msg_t reply;
	char *buf, *func;
	struct ipc_obj *obj;
	void *vdata;
	GString *domain = g_string_new("");

	GVariant *var, *var_par, *ret, *ret_norm, *ret_var;

	void *context = zmq_ctx_new ();
	void *responder = zmq_socket (context, ZMQ_REP);

	g_string_printf(domain, "%s%s.%s", IPC_PREFIX, daemon_name, "ipc");
	
	zmq_bind (responder, domain->str);
	g_string_free(domain, TRUE);

	for(;;) {
		G_LOCK(ipc_handler);
		zmq_msg_init (&request);
		zmq_msg_recv (&request, responder, 0);

		if (exiting) {
			zmq_close (responder);
			return NULL;
		}
		vdata = malloc(zmq_msg_size(&request));
		memcpy(vdata, zmq_msg_data(&request), zmq_msg_size(&request));
		var = g_variant_new_from_data(((const GVariantType *) "(sv)"), 
			vdata, zmq_msg_size(&request), TRUE, NULL, NULL);
		zmq_msg_close (&request);

		g_variant_get(var, "(&sv)", &func, &var_par);
		obj = g_datalist_get_data(&fmap, func);/*XXX error handle*/
		ret =  obj->callback(var_par);
		ret_var = g_variant_new_variant(ret);/*Don't have to free ret if ret_var are free'd*/

		ret_norm = g_variant_get_normal_form(ret_var);

		//  Send reply back to client
		zmq_msg_init_size (&reply, g_variant_get_size(ret_norm));
		memcpy (zmq_msg_data (&reply), g_variant_get_data(ret_norm), g_variant_get_size(ret_norm));
		zmq_msg_send (&reply, responder, 0);
		zmq_msg_close (&reply);

		/*Free*/
		g_variant_unref(ret_norm);
		g_variant_unref(ret_var);
		g_variant_unref(var_par);
		g_variant_unref(var);
		G_UNLOCK(ipc_handler);

	}
}

void ipc_add(char *func, GVariant *callback(GVariant *par), char *in, char *out){
	struct ipc_obj *obj = malloc(sizeof(struct ipc_obj));
	obj->func = g_strdup(func);
	obj->in = g_strdup(in);
	obj->out = g_strdup(out);
	obj->callback = callback;
	if (g_datalist_get_data(&fmap, func) == NULL) {
		g_datalist_set_data(&fmap, func, obj);
	}		

}

void socket_free(GQuark key, gpointer data, gpointer user_data)
{
	zmq_close(data);
}

void ipc_release(void)
{
	exiting = 1;
	g_datalist_foreach(&dmap, socket_free, NULL);
	zmq_term(_ipc_context);
	if (_ipc_context)
		zmq_ctx_destroy (_ipc_context);
}

void ipc_signal_dispatch(gpointer data, gpointer user_data) {
	GVariant *par = data;
	GVariant *var;
	char *daemon, *func;
	void (*signal_callback)(GVariant *par);
	GString *str = g_string_new("");

	g_variant_get(par, "(&s&sv)", &daemon, &func, &var);
	g_string_printf(str, "%s %s", daemon, func);

	signal_callback = g_datalist_get_data(&smap, str->str);
	if (signal_callback) {
		signal_callback(var);
	}
	g_string_free(str, TRUE);
	g_variant_unref(par);
	//printf("g_thread_pool_get_num_threads : %d\n", g_thread_pool_get_num_threads(sig_pool));
	//g_variant_unref(var);

}

GVariant *ipc_signal_handler(GVariant *par) {
	g_variant_ref(par);
	g_thread_pool_push(sig_pool, par, NULL);
	//ipc_signal_dispatch((gpointer)par, NULL);
	return g_variant_new("()");
}

void ipc_init(char *name){
	daemon_name = g_strdup(name);
	_ipc_context = zmq_ctx_new ();
	
	if (name == NULL) {
		return;
	}
	//_responder = zmq_socket (_ipc_context, ZMQ_REP);

	//zmq_bind (_responder, domain->str);
	g_datalist_init(&fmap);
	g_datalist_init(&dmap);
	g_datalist_init(&smap);
	g_thread_new("ipc handler thread", ipc_handler, NULL);
	sig_pool = g_thread_pool_new(ipc_signal_dispatch, NULL, -1, FALSE, NULL);

	/*Add signal hander*/
	ipc_add("_Sig", ipc_signal_handler, "(ssv)", "()");
}


void ipc_raise(char *func, GVariant *par){
	GVariant *ret;
	GString *str = g_string_new("");
	GVariant *request = g_variant_new("(ssv)", daemon_name, func, par);
	ret = ipc_call("libcpcd", "RaiseSignal", request);
	g_variant_unref(ret);
}

void ipc_signal(char *daemon, char *func, void callback(GVariant *par)){
	GString *str = g_string_new("");
	GVariant *request = g_variant_new("(sss)", daemon_name, daemon, func);
	GVariant *ret;
	g_string_printf(str, "%s %s", daemon, func);

	ret = ipc_call("libcpcd", "RegSignal", request);
	if ( ret && (g_datalist_get_data(&smap, str->str) == NULL) ) {
		g_datalist_set_data(&smap, str->str, callback);
	}
	g_string_free(str, TRUE);
}

G_LOCK_DEFINE(ipc_call);
GVariant *ipc_call(char *daemon, char *func, GVariant *par){

	GString *domain = g_string_new("");
	GVariant *var, *var_norm, *ret, *ret_var;
	gsize var_size_norm;
	zmq_msg_t request;
	zmq_msg_t reply;
	void *requester;
	void *data;

	g_string_printf(domain, "%s%s.%s", IPC_PREFIX, daemon, "ipc");

	if ( (requester = g_datalist_get_data(&dmap, daemon)) == NULL) {
		requester = zmq_socket (_ipc_context, ZMQ_REQ);
		zmq_connect (requester, domain->str);
		g_datalist_set_data(&dmap, daemon, requester);
	}

	G_LOCK(ipc_call);
	var = g_variant_new("(sv)", func, par);
	var_norm = g_variant_get_normal_form(var);
	var_size_norm = g_variant_get_size(var_norm);
	zmq_msg_init_size (&request, var_size_norm);
	g_variant_store(var_norm, zmq_msg_data (&request));
	zmq_msg_send (&request, requester, 0);
	zmq_msg_close (&request);

	zmq_msg_init (&reply);
	zmq_msg_recv (&reply, requester, 0);
	data = malloc(zmq_msg_size(&reply));
	memcpy(data, zmq_msg_data(&reply), zmq_msg_size(&reply));
	ret_var = g_variant_new_from_data(G_VARIANT_TYPE_VARIANT, data, 
		zmq_msg_size(&reply), TRUE, NULL, NULL);

	ret = g_variant_get_variant(ret_var);

	zmq_msg_close (&reply);
	g_variant_unref(var);
	g_variant_unref(var_norm);
	G_UNLOCK(ipc_call);


	return ret;

}
